package com.mjf.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/**
 * Kafka 工具类
 */
public class MyKafkaUtil {

    private static String BROKERS = "hadoop102:9092,hadoop103:9092,hadoop104:9092";
    private static String DEFAULT_TOPIC = "DWD_DEFAULT_TOPIC";

    /**
     * 生成 Kafka 生产者
     *
     * @param topic 主题
     * @return FlinkKafka 生产者
     */
    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        return new FlinkKafkaProducer<String>(
                BROKERS,
                topic,
                new SimpleStringSchema()
        );
    }

    public static <T> FlinkKafkaProducer<T> getKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERS);

        return new FlinkKafkaProducer<T>(
                DEFAULT_TOPIC,
                kafkaSerializationSchema,
                prop,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE    // 需要 checkpoint 打开才能生效，会报错
        );
    }

    /**
     * 生成 Kafka 消费者
     *
     * @param topic   主题
     * @param groupId 消费者组
     * @return FlinkKafka 消费者
     */
    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
        Properties prop = new Properties();
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERS);

        return new FlinkKafkaConsumer<String>(
                topic,
                new SimpleStringSchema(),
                prop
        );
    }

    /**
     * 拼接 Kafka 相关属性到 DDL
     *
     * @param topic 主题
     * @param groupId 消费者组
     * @return DDL描述
     */
    public static String getKafkaDDL(String topic, String groupId) {
        return  " 'connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + BROKERS + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'latest-offset'  ";
    }

}
